package com.ferry.id.pool;

import com.ferry.id.common.IdGenetator;
import com.ferry.id.common.NamedThreadFactory;
import com.ferry.id.common.Result;
import com.ferry.id.common.Status;
import com.ferry.id.pool.dao.AllocDao;
import com.ferry.id.pool.domain.Alloc;
import com.ferry.id.pool.domain.Pool;
import com.ferry.id.pool.domain.PooledBuffer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;

/**
 * Created by daiyong
 */
public class PoolIdGenerator implements IdGenetator {

	private static final Logger logger = LoggerFactory.getLogger(PoolIdGenerator.class);

	private AllocDao allocDao;

	private volatile boolean init = false;

	private Map<String, PooledBuffer> cache = new ConcurrentHashMap<String, PooledBuffer>();

	/**
	 * 最大步长
	 */
	private static final int MAX_STEP = 1000000;

	/**
	 * 每个池子保持15分钟
	 */
	private static final long POOL_DURATION = 15 * 60 * 1000;

	/**
	 * 线程池
	 */
	private ExecutorService service =
			new ThreadPoolExecutor(5, Integer.MAX_VALUE, 60L, TimeUnit.SECONDS, new SynchronousQueue<Runnable>(), new NamedThreadFactory(""));

	@Override
	public Result getId(String namespace) {

		if (!init) {
			return new Result(-1, Status.FAIL);
		}

		if (cache.containsKey(namespace)) {
			PooledBuffer pooledBuffer = cache.get(namespace);
			/**
			 * 没有初始化完成则加载pool
			 */
			if (!pooledBuffer.isInitOk()) {
				//使用 pooledBuffer 可以缩小锁粒度
				synchronized (pooledBuffer) {
					//并且使用双重检查
					if (!pooledBuffer.isInitOk()) {
						try {
							loadPoolFromDb(namespace, pooledBuffer.getCurrentPool());
							pooledBuffer.setInitOk(true);
						} catch (Exception e) {
							logger.error("load pool from db exception, namespace:{}", namespace, e);
						}
					}
				}
			}

			return getIdFromPool(cache.get(namespace));
		}

		return new Result(-3, Status.FAIL);
	}

	/**
	 * 真正的生成id
	 * @param pooledBuffer
	 * @return
	 */
	private Result getIdFromPool(final PooledBuffer pooledBuffer) {

		//死循环获取
		while (true) {
			try {
				pooledBuffer.rLock().lock();

				//正在使用的pool
				final Pool currentPool = pooledBuffer.getCurrentPool();
				//异步加载另一个pool
				if (!pooledBuffer.isNextReady() && (currentPool.getIdle() <= 0.9 * currentPool.getStep()) && pooledBuffer.getThreadRunning().compareAndSet(false, true)) {
					service.execute(new Runnable() {
						@Override
						public void run() {
							//下一个pool
							Pool nextPool = pooledBuffer.getPools()[pooledBuffer.nextPos()];
							boolean updateOk = false;
							try {
								loadPoolFromDb(pooledBuffer.getNamespace(), nextPool);
								updateOk = true;
							} catch (Exception e) {
								logger.error("loadPoolFromDb Exception, namespace{}", pooledBuffer.getNamespace(), e);
							} finally {
								if (updateOk) {
									pooledBuffer.wLock().lock();
									pooledBuffer.setNextReady(true);
									pooledBuffer.getThreadRunning().set(false);
									pooledBuffer.wLock().unlock();
								} else {
									pooledBuffer.getThreadRunning().set(false);
								}
							}
						}
					});
				}

				long value = currentPool.getValue().getAndIncrement();
				if (value < currentPool.getMaxId()) {
					return new Result(value, Status.SUCCESS);
				}
			} finally {
				pooledBuffer.rLock().unlock();
			}

			//等待异步线程加载完毕 自旋
			waitPooledBuffer(pooledBuffer);

			//切换另一个pool
			try {
				pooledBuffer.wLock().lock();

				final Pool currentPool = pooledBuffer.getCurrentPool();

				long value = currentPool.getValue().getAndIncrement();
				if (value < currentPool.getMaxId()) {
					return new Result(value, Status.SUCCESS);
				}

				if (pooledBuffer.isNextReady()) {
					pooledBuffer.switchPos();
					pooledBuffer.setNextReady(false);
				}
				else {
					logger.error("Both two segments in {} are not ready!", pooledBuffer);
					return new Result(-3, Status.FAIL);
				}

			} finally {
				pooledBuffer.wLock().unlock();
			}
		}


	}

	private void waitPooledBuffer(PooledBuffer pooledBuffer) {
		int roll = 0;
		while (pooledBuffer.getThreadRunning().get()) {
			roll += 1;
			if(roll > 10000) {
				try {
					TimeUnit.MILLISECONDS.sleep(10);
					break;
				} catch (InterruptedException e) {
					logger.warn("Thread {} Interrupted",Thread.currentThread().getName());
					break;
				}
			}
		}
	}

	/**
	 * 从db中加载pool
	 * @param namespace
	 * @param currentPool
	 */
	private void loadPoolFromDb(String namespace, Pool currentPool) {

		/**
		 * 首先设置 pooledBuffer
		 */
		PooledBuffer pooledBuffer = currentPool.getPooledBuffer();
		Alloc alloc;

		//第一次进入
		if (!pooledBuffer.isInitOk()) {
			alloc = allocDao.updateMaxIdAndGetAlloc(namespace);
			pooledBuffer.setStep(alloc.getStep());
			pooledBuffer.setMinStep(alloc.getStep());
		}
		//第二次进入
		else if (pooledBuffer.getUpdateTimestamp() == 0) {
			alloc = allocDao.updateMaxIdAndGetAlloc(namespace);
			pooledBuffer.setUpdateTimestamp(System.currentTimeMillis());
			pooledBuffer.setMinStep(alloc.getStep());
		}
		//第三次及以后进入 动态设置步长
		else {
			long duration = System.currentTimeMillis() - pooledBuffer.getUpdateTimestamp();
			int nextStep = pooledBuffer.getStep();

			if (duration < POOL_DURATION) {
				if (nextStep * 2 > MAX_STEP) {
					//do nothing
				} else {
					// 步数 * 2
					nextStep = nextStep * 2;
				}
			}
			// 15分钟 < duration < 30分钟 : nothing
			else if (duration < POOL_DURATION * 2) {
				//do nothing with nextStep
			}
			// duration > 30 分钟 : 缩小step ,最小为DB中配置的步数
			else {
				nextStep = nextStep / 2 >= pooledBuffer.getMinStep() ? nextStep / 2 : nextStep;
			}

			//保存到数据库
			Alloc tmpAlloc = new Alloc();
			tmpAlloc.setNamespace(namespace);
			tmpAlloc.setStep(nextStep);
			tmpAlloc.setUpdateTime(System.currentTimeMillis());
			alloc = allocDao.updateMaxIdByCustomStepAndGetLeafAlloc(tmpAlloc);

			pooledBuffer.setUpdateTimestamp(System.currentTimeMillis());
			pooledBuffer.setStep(nextStep);
			pooledBuffer.setMinStep(alloc.getStep());
		}

		/**
		 * 设置pool
		 */
		long value = alloc.getMaxId() - pooledBuffer.getStep();  //初始值
		currentPool.getValue().set(value);
		currentPool.setMaxId(alloc.getMaxId());
		currentPool.setStep(pooledBuffer.getStep());
	}

	@Override
	public synchronized boolean init() {

		if (init) {
			logger.warn("PoolIdGenerator has inited");
			return true;
		}

		logger.info("PoolIdGenerator init...");

		loadDataFromDb();

		init = true;

		loadDataFromDbAsync();

		return init;
	}

	/**
	 * 每隔一分钟 周期性加载db数据
	 */
	private void loadDataFromDbAsync() {
		ScheduledExecutorService service = Executors.newSingleThreadScheduledExecutor(new NamedThreadFactory("ferry-id-PoolIdGenerator"));
		service.scheduleWithFixedDelay(new Runnable() {
			@Override
			public void run() {
				loadDataFromDb();
			}
		}, 60, 60, TimeUnit.SECONDS);
	}

	private void loadDataFromDb() {
		logger.info("start loadDataFromDb...");

		List<String> dbNamespaces = allocDao.getAllNamespaces();
		if (null == dbNamespaces || dbNamespaces.isEmpty()) {
			logger.warn("have no dbNamespaces");
		}
		else {
			List<String> cachedNamespaces = new ArrayList<String>(cache.keySet());
			Set<String> insertNamespacesSet = new HashSet<String>(dbNamespaces);
			Set<String> removeNamespacesSet = new HashSet<String>(cachedNamespaces);

			//过滤出应该新增的namespace
			for(int i = 0; i < cachedNamespaces.size(); i++){
				String namespace = cachedNamespaces.get(i);
				if (insertNamespacesSet.contains(namespace)) {
					insertNamespacesSet.remove(namespace);
				}
			}
			for (String namespace : insertNamespacesSet) {
				PooledBuffer pooledBuffer = new PooledBuffer();
				pooledBuffer.setNamespace(namespace);
				Pool pool = pooledBuffer.getCurrentPool();
				pool.setValue(new AtomicLong(0));
				pool.setMaxId(0);
				pool.setStep(0);
				cache.put(namespace, pooledBuffer);
				logger.info("Add namespace {} from db to IdCache, PooledBuffer {}", namespace, pooledBuffer);
			}

			//删除过期的namespace
			for(int i = 0; i < dbNamespaces.size(); i++){
				String tmp = dbNamespaces.get(i);
				if(removeNamespacesSet.contains(tmp)){
					removeNamespacesSet.remove(tmp);
				}
			}
			for (String namespace : removeNamespacesSet) {
				cache.remove(namespace);
				logger.info("Remove namespace {} from IdCache", namespace);
			}

		}

		logger.info("finish loadDataFromDb...");
	}

	public void setAllocDao(AllocDao allocDao) {
		this.allocDao = allocDao;
	}
}
